package io.activej.cube;

import io.activej.cube.CubeStructure.AggregationConfig;
import io.activej.cube.aggregation.AggregationChunk;
import io.activej.cube.aggregation.PrimaryKey;
import io.activej.cube.aggregation.ot.AggregationDiff;
import io.activej.cube.aggregation.predicate.AggregationPredicate;
import io.activej.cube.aggregation.predicate.AggregationPredicates;
import io.activej.cube.ot.CubeDiff;
import io.activej.etl.LogDiff;
import io.activej.etl.LogState;
import io.activej.fs.FileSystem;
import io.activej.ot.StateManager;
import org.junit.Before;
import org.junit.Test;

import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDate;
import java.time.Month;
import java.util.*;

import static io.activej.cube.CubeStructure.AggregationConfig.id;
import static io.activej.cube.aggregation.PrimaryKey.ofArray;
import static io.activej.cube.aggregation.fieldtype.FieldTypes.*;
import static io.activej.cube.aggregation.measure.Measures.sum;
import static io.activej.cube.aggregation.predicate.AggregationPredicates.gt;
import static io.activej.cube.aggregation.predicate.AggregationPredicates.notEq;
import static io.activej.promise.TestUtils.await;
import static java.util.stream.Collectors.toSet;
import static org.junit.Assert.assertEquals;

public final class CubeGetIrrelevantChunksTest extends CubeTestBase {
	private static final int NUMBER_MIN = 0;
	private static final int NUMBER_MAX = 100;

	private static final int DATE_MIN_DAYS = (int) LocalDate.of(2020, Month.JANUARY, 1).toEpochDay();
	private static final int DATE_MAX_DAYS = (int) LocalDate.of(2021, Month.JANUARY, 1).toEpochDay();

	private static final LocalDate LOWER_DATE_BOUNDARY = LocalDate.of(2020, Month.JULY, 31);
	private static final int LOWER_DATE_BOUNDARY_DAYS = (int) LOWER_DATE_BOUNDARY.toEpochDay();
	private static final AggregationPredicate DATE_PREDICATE = gt("date", LOWER_DATE_BOUNDARY);
	private static final int LOWER_NUMBER_BOUNDARY = 50;
	private static final AggregationPredicate ADVERTISER_PREDICATE = gt("advertiser", LOWER_NUMBER_BOUNDARY);

	private AggregationConfig dateAggregation;
	private AggregationConfig advertiserDateAggregation;
	private AggregationConfig enumAggregation;
	private StateManager<LogDiff<CubeDiff>, LogState<CubeDiff, CubeState>> stateManager;
	private CubeStructure cubeStructure;

	private long chunkId;

	private final Set<Object> toBePreserved = new HashSet<>();
	private final Set<Object> toBeCleanedUp = new HashSet<>();

	private List<LogDiff<CubeDiff>> diffs;

	@Before
	public void before() throws Exception {
		diffs = new ArrayList<>();
		chunkId = 0;
		toBeCleanedUp.clear();
		toBePreserved.clear();
		Path aggregationsDir = temporaryFolder.newFolder().toPath();

		FileSystem fs = FileSystem.builder(reactor, EXECUTOR, aggregationsDir)
			.withTempDir(Files.createTempDirectory(""))
			.build();
		await(fs.start());

		dateAggregation = id("date")
			.withDimensions("date")
			.withMeasures("impressions", "clicks", "conversions", "revenue");

		advertiserDateAggregation = id("advertiser-date")
			.withDimensions("advertiser", "date")
			.withMeasures("impressions", "clicks", "conversions", "revenue");

		enumAggregation = id("enum-date")
			.withDimensions("enum", "date")
			.withMeasures("impressions", "clicks", "conversions", "revenue");

		CubeStructure basicCubeStructure = builderOfBasicCubeStructure()
			.withAggregation(dateAggregation)
			.withAggregation(advertiserDateAggregation)
			.withAggregation(enumAggregation)
			.build();

		stateManager = stateManagerFactory.create(basicCubeStructure, description);
	}

	@Test
	public void date() throws Exception {
		cubeStructure = builderOfBasicCubeStructure()
			.withAggregation(dateAggregation.withPredicate(DATE_PREDICATE))
			.build();

		toBePreserved.add(addChunk("date", ofArray(DATE_MIN_DAYS), ofArray(DATE_MAX_DAYS)));
		toBePreserved.add(addChunk("date", ofArray(DATE_MIN_DAYS + 50), ofArray(DATE_MAX_DAYS - 50)));

		toBeCleanedUp.add(addChunk("date", ofArray(DATE_MIN_DAYS), ofArray(LOWER_DATE_BOUNDARY_DAYS)));
		toBeCleanedUp.add(addChunk("date", ofArray(DATE_MIN_DAYS), ofArray(DATE_MIN_DAYS + 50)));

		doTest();
	}

	@Test
	public void anEnum() {
		cubeStructure = builderOfBasicCubeStructure()
			.withAggregation(enumAggregation.withPredicate(notEq("enum", null)))
			.build();

		toBePreserved.add(addChunk("enum-date", ofArray(MyEnum.ONE, DATE_MIN_DAYS), ofArray(MyEnum.ONE, DATE_MAX_DAYS)));
		toBePreserved.add(addChunk("enum-date", ofArray(MyEnum.ONE, DATE_MIN_DAYS + 50), ofArray(MyEnum.TWO, DATE_MAX_DAYS - 50)));

		doTest();
	}

	@Test
	public void advertiserDate_DatePredicate() throws Exception {
		cubeStructure = builderOfBasicCubeStructure()
			.withAggregation(advertiserDateAggregation.withPredicate(DATE_PREDICATE))
			.build();
		stateManager = stateManagerFactory.create(cubeStructure, description);

		toBePreserved.add(addChunk("advertiser-date", ofArray(5, DATE_MIN_DAYS), ofArray(6, DATE_MAX_DAYS)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(1, DATE_MIN_DAYS + 50), ofArray(20, DATE_MAX_DAYS - 50)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(6, DATE_MIN_DAYS), ofArray(7, DATE_MIN_DAYS + 1)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(10, DATE_MIN_DAYS), ofArray(10, LOWER_DATE_BOUNDARY_DAYS + 1)));

		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(5, DATE_MIN_DAYS), ofArray(5, DATE_MIN_DAYS + 50)));
		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(10, DATE_MIN_DAYS + 50), ofArray(10, LOWER_DATE_BOUNDARY_DAYS)));

		doTest();
	}

	@Test
	public void advertiserDate_AdvertiserPredicate() throws Exception {
		cubeStructure = builderOfBasicCubeStructure()
			.withAggregation(advertiserDateAggregation.withPredicate(ADVERTISER_PREDICATE))
			.build();
		stateManager = stateManagerFactory.create(cubeStructure, description);

		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MAX_DAYS)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN + 50, DATE_MIN_DAYS + 50), ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 1, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MIN_DAYS + 1)));

		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY, DATE_MIN_DAYS + 50)));
		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 10, DATE_MIN_DAYS + 50), ofArray(LOWER_NUMBER_BOUNDARY - 5, LOWER_DATE_BOUNDARY_DAYS)));

		doTest();
	}

	@Test
	public void advertiserDate_AdvertiserPredicateAndDatePredicate() throws Exception {
		cubeStructure = builderOfBasicCubeStructure()
			.withAggregation(advertiserDateAggregation
				.withPredicate(AggregationPredicates.and(ADVERTISER_PREDICATE, DATE_PREDICATE)))
			.build();
		stateManager = stateManagerFactory.create(cubeStructure, description);

		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MAX_DAYS)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN + 50, DATE_MIN_DAYS + 50), ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 1, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MIN_DAYS + 1)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50), ofArray(NUMBER_MAX - 5, DATE_MAX_DAYS - 25)));
		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50), ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 25)));

		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY, DATE_MAX_DAYS - 50)));
		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 10, DATE_MIN_DAYS + 50), ofArray(LOWER_NUMBER_BOUNDARY - 5, LOWER_DATE_BOUNDARY_DAYS + 10)));
		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 10, DATE_MIN_DAYS + 50), ofArray(LOWER_NUMBER_BOUNDARY - 10, LOWER_DATE_BOUNDARY_DAYS + 10)));
		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(NUMBER_MIN + 10, DATE_MAX_DAYS - 50), ofArray(NUMBER_MIN + 10, DATE_MAX_DAYS - 25)));

		doTest();
	}

	private void doTest() {
		await(stateManager.push(diffs));

		Set<Object> expectedChunks = new HashSet<>();
		expectedChunks.addAll(toBePreserved);
		expectedChunks.addAll(toBeCleanedUp);

		assertEquals(expectedChunks, stateManager.query(logState -> logState.getDataState().getAllChunks()));

		stateManager = stateManagerFactory.createUninitialized(cubeStructure, description);
		stateManagerFactory.start(stateManager);

		Set<Object> irrelevantChunks = stateManager.query(logState -> logState.getDataState().getIrrelevantChunks())
			.values()
			.stream()
			.flatMap(Collection::stream)
			.map(AggregationChunk::getChunkId)
			.collect(toSet());
		assertEquals(toBeCleanedUp, irrelevantChunks);
	}

	private long addChunk(String aggregationId, PrimaryKey minKey, PrimaryKey maxKey) {
		long chunkId = ++this.chunkId;
		diffs.add(LogDiff.forCurrentPosition(
			CubeDiff.of(Map.of(
				aggregationId, AggregationDiff.of(Set.of(
					AggregationChunk.create(
						chunkId,
						cubeStructure.getAggregationStructure(aggregationId).getMeasures(),
						minKey,
						maxKey,
						1))))
			)));
		return chunkId;
	}

	private CubeStructure.Builder builderOfBasicCubeStructure() {
		return CubeStructure.builder()
			.withDimension("date", ofLocalDate())
			.withDimension("advertiser", ofInt())
			.withDimension("campaign", ofInt())
			.withDimension("banner", ofInt())
			.withDimension("enum", ofEnum(MyEnum.class))
			.withRelation("campaign", "advertiser")
			.withRelation("banner", "campaign")
			.withMeasure("impressions", sum(ofLong()))
			.withMeasure("clicks", sum(ofLong()))
			.withMeasure("conversions", sum(ofLong()))
			.withMeasure("revenue", sum(ofDouble()));
	}

	public enum MyEnum {
		ONE, TWO, THREE
	}
}
